package com.atguigu.flink.watermark;

import com.atguigu.flink.function.MyUtil;
import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;

/**
 * Created by Smexy on 2023/2/28
 */
public class Demo4_CustomWMStrategy
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            //自定义水印策略
            .<WaterSensor>forGenerator(new WatermarkGeneratorSupplier<WaterSensor>()
            {
                //返回一个  WatermarkGenerator
                @Override
                public WatermarkGenerator<WaterSensor> createWatermarkGenerator(Context context) {
                    return new MyWatermarkGenerator(3000);
                }
            })
            .withTimestampAssigner((e, ts) -> e.getTs());

                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .assignTimestampsAndWatermarks(watermarkStrategy)
                   .keyBy(WaterSensor::getId)
                   .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                   .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>()
                   {
                       @Override
                       public void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                           List<WaterSensor> waterSensors = MyUtil.parseToList(elements);
                           out.collect(key + ":" + context.window() + " :" + waterSensors);
                       }
                   })
                   .print();


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    private static class MyWatermarkGenerator implements WatermarkGenerator<WaterSensor>{

        private long delay;

        public MyWatermarkGenerator(long delayms){
            delay = delayms;
        }
        //保证水印不能回退，必须是单调递增。
        private long maxTimestamp = Long.MIN_VALUE;
        //水印一定 <= 事件时间。否则就会造成数据大面积迟到，无法计算。

        //发送水印，断点式
        @Override
        public void onEvent(WaterSensor event, long eventTimestamp, WatermarkOutput output) {
            maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
            Watermark watermark = new Watermark(maxTimestamp - delay);
            System.out.println("向下游发送wm:"+watermark.getTimestamp());
            output.emitWatermark(watermark);
        }

        //发送水印，周期性
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {

        }
    }
}
